package com.epoint.flinkdemo.utils;

import com.epoint.flinkdemo.annotation.TableField;
import com.epoint.flinkdemo.annotation.TableName;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.text.WordUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.annotation.Annotation;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * @author liufl
 * @version [版本号, 21-4-25]
 */
public class RowUtils
{

    private static final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private static final Logger log = LoggerFactory.getLogger(RowUtils.class);

    /**
     * ROW对象转尸体
     * @param row
     * @param t
     * @param <T>
     * @return
     */
    public static <T> T convertToEntity(Row row, Class<T> t) {
        Annotation annotation = t.getAnnotation(TableName.class);
        Map<String, Object> rowMap = (Map<String, Object>)row.getField(0);
        if (annotation == null) {
            return null;
        }
        T ins = null;
        try {
            Field[] fields = t.getDeclaredFields();
            ins = t.newInstance();
            for (Field field : fields) {
                String fieldName = field.getName();
                String key = "";
                Class<?> fieldType = field.getType();
                TableField tableField = field.getAnnotation(TableField.class);
                if (tableField != null) {
                    key = "after_" + tableField.value();

                    Object obj = rowMap.get(key);
                    Method method = t.getMethod("set" + WordUtils.capitalize(fieldName), new Class[]{fieldType});
                    if (obj != null && method != null) {
                        if (fieldType.equals(String.class)) {
                            method.invoke(ins, obj.toString());
                        }
                        else if (fieldType.equals(Integer.class)) {
                            if (StringUtils.isNotBlank(obj.toString())) {
                                method.invoke(ins, Integer.parseInt(obj.toString()));
                            }
                        }
                        else if (fieldType.equals(Long.class)) {
                            if (StringUtils.isNotBlank(obj.toString())) {
                                method.invoke(ins, Long.parseLong(obj.toString()));
                            }
                        }
                        else if (fieldType.equals(Date.class)) {
                            if (StringUtils.isNotBlank(obj.toString())) {
                                Date date = dateFormat.parse(obj.toString());
                                method.invoke(ins, date);
                            }
                        }
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return ins;
    }

    /**
     * 对象转sql
     * @param t
     * @return
     */
    public static <T> String toInsertSql(T t) {
        StringBuilder sql = new StringBuilder("insert into ");
        Class<?> tClass = t.getClass();
        TableName tableName = tClass.getAnnotation(TableName.class);
        sql.append(tableName.value() + "(");

        Field[] fields = tClass.getDeclaredFields();
        List<Tuple2<String, Object>> tuple2List = new ArrayList<>();
        try {
            for (Field field : fields) {
                TableField tableField = field.getAnnotation(TableField.class);
                if (tableField != null) {
                    Tuple2<String, Object> tuple2 = new Tuple2<>();
                    tuple2.f0 = tableField.value();
                    Method method = tClass.getMethod("get" + WordUtils.capitalize(field.getName()), null);
                    Object result = method.invoke(t, null);
                    if (result instanceof Date) {
                        tuple2.f1 = dateFormat.format(result);
                    }
                    else {
                        tuple2.f1 = result;
                    }

                    tuple2List.add(tuple2);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        List<String> fieldList = new ArrayList<>();
        List<Object> valueList = new ArrayList<>();
        for (Tuple2<String, Object> tuple2 : tuple2List) {
            fieldList.add(tuple2.f0);
            if (tuple2.f1 == null) {
                valueList.add("null");
            }
            else if (tuple2.f1 instanceof Integer || tuple2.f1 instanceof Long) {
                valueList.add(tuple2.f1);
            }
            else if (tuple2.f1 instanceof String) {
                valueList.add("'" + tuple2.f1 + "'");
            }
        }
        sql.append(StringUtils.join(fieldList, ",") + ") values(" + StringUtils.join(valueList, ",") + ")");
        log.info("ck.sql = " + sql.toString());
        return sql.toString();
    }
}
